package spark.sparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * scala 的窗口操作
  */
object ScalaWindow{
	def main(args: Array[String]): Unit = {
		// scala 中的窗口操作
		val ints = Array(1,2,3,4)
		val intses: Iterator[Array[Int]] = ints.sliding(2,2)
		for(list <- intses){
			println(list.mkString(",")) // 1,2/3,4
		}
		println(ints.slice(1,3).toList) // List(2, 3)
	}
}


class SpakStreamingWindow {
	def main(args: Array[String]): Unit = {
		val conf: SparkConf = new SparkConf().setAppName("SpakStreamingWindow").setMaster("local[*]")
		val streamingContext = new StreamingContext(conf, Seconds(3))
		
		val socketLineDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("nn1.hadoop", 9999)
		socketLineDStream.map{
			case line => {
				// TODO 代码，（Executor）
				line
			}
		}
		
		// TODO 代码，（Driver），只执行一次
//		socketLineDStream.transform{
//			case rdd => {
//				// TODO 代码，（Driver）,每个批次执行一次，可以动态更新
//				val resRDD: RDD[Nothing] = rdd.map {
//					case line => {
//						// TODO 代码，（Executor），每条数据执行一次
//						line
//					}
//				}
//				resRDD
//			}
//		}
		
		// 一个采集周期有一个RDD
		socketLineDStream.foreachRDD(rdd => {
			rdd.foreach(println)
		})
		
		// 一个采集周期有一个RDD
		socketLineDStream.foreachRDD(rdd => {
			rdd.foreachPartition{
				// 数据库的连接每个partition创建一次
				case iter => iter
			}
		})
		
	}
}
